package com.cti.commons.delayqueue.handler;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.cti.commons.delayqueue.core.DelayQueue;
import com.cti.commons.delayqueue.core.mode.DelayMessage;
import com.cti.commons.delayqueue.repository.DelayConsumerMessageLog;
import com.cti.commons.delayqueue.repository.DelayConsumerMessageRepository;

import cn.hutool.extra.spring.SpringUtil;

/**
 * @author et.wang
 * @description: 扫描延迟任务桶中的任务，放到准备队列中
 * @date 2022/10/27 19:35
 */
@Component
public class DelayQueueHandler {

    public static final int SIZE = 1024;

    private static final Logger logger = LoggerFactory.getLogger(DelayQueueHandler.class);

    @Autowired
    private DelayQueue delayQueue;

    @Autowired
    private DelayConsumerMessageRepository repository;

    //@SchedulerLock(name = "${delay.queue.queueJobName:delayQueueHandler}",lockAtMostFor = "PT6s")
    public void execute() {
        DelayMessage delayMessage = null;
        DelayConsumerMessageLog log = null;
        try {
            List<DelayQueueListenerHandler> delayQueueListenerHandler =
                SpringUtil.getBean("delayQueueListenerHandler", List.class);
            for (DelayQueueListenerHandler delayMessageHandler : delayQueueListenerHandler) {
                delayMessage = delayQueue.pop(delayMessageHandler.getTopic());
                if (delayMessage == null) {
                    continue;
                }
                log = new DelayConsumerMessageLog(delayMessage.getId() + "",
                    delayMessage.getTopic(),
                    delayMessage.toString(),
                    null);
                boolean execute = delayMessageHandler.handleMessage(delayMessage);
                if (execute) {
                    delayQueue.finish(delayMessage.getId());
                    log.success();
                    logger.info("=================延时队列消费处理 [{}] 执行成功=======================",
                        delayMessage.getId());
                }
                repository.save(log);
            }
        } catch (Exception e) {
            logger.error("消费出错id：{}", delayMessage.getId(), e);
            String errorMessageString = e.getMessage();
            if (log == null) {
                log = new DelayConsumerMessageLog(errorMessageString);
            }
            log.setException(errorMessageString.length() < SIZE ? errorMessageString :
                errorMessageString.substring(0, SIZE));
            log.fail();
            repository.save(log);
        } finally {
            if (delayMessage != null) {
                delayQueue.finish(delayMessage.getId());
            }
        }
    }

}
